﻿#include <stdlib.h>
#include <stdio.h>
#include <iostream>
#include <pthread.h>
#include <semaphore.h>
#include <string>
#include <iostream>
#include <algorithm>
#include <fstream>
#include <mutex>
using namespace std;

#define P 8 //线程数
int C = 2; //用于启动循环
double alpha = 0.85;
double elp = 1e-20; //判断收敛阈值

const char filePath[50] = "dataset\\Data.txt";

int N = 0;
int nodeN = 0;
int** nodeOutDegreesP;
int* nodeOutDegrees; //统计各个节点的出度

typedef struct {
    int nodeId;
    double prVal;
}prRes; //仅用于展示pagerank排序结果

typedef struct {
    int r;
    int c;
    double val;
}coo;

typedef struct {
    int p_id; //线程id
    coo* cooM; //矩阵M
    double* pr; //列向量pagerank
    double* updatedPr; //每一轮更新后pr（防止多线程脏读）
    int** nodeOutDegreesP; //同样防止脏读
}threadParam_t;

//查看矩阵M的值
void displayM(coo* cooM) {
    cout << "N = " << N << endl;
    for (int i = 0; i < N; i++) {
        cout << "[" << cooM[i].r << "][" << cooM[i].c << "] = " << cooM[i].val << endl;
    }
    return;
}

//查看某个节点的入度
int getInDegree(coo* cooM, int toNodeId) {
    int num = 0;
    for (int i = 0; i < N; i++) {
        if (cooM[i].r == toNodeId) {
            num++;
        }
    }
    return num;
}

bool cmp(prRes a, prRes b) {
    return a.prVal > b.prVal;
}

//查看列向量pr的值（前n个）
void displayPr(prRes* result, int n = 100) {
    sort(result, result + nodeN, cmp);
    for (int i = 0; i < n; i++) {
        cout << result[i].nodeId << " " << result[i].prVal << endl;
    }
}

//保存结果
void saveResult(prRes* result, int n = 100) {
    ofstream ofs;
    ofs.open("output\\pagerankResult.txt", ios::out);
    for (int i = 0; i < n - 1; i++) {
        ofs << result[i].nodeId << " " << result[i].prVal << endl;
    }
    ofs << result[n - 1].nodeId << " " << result[n - 1].prVal;
    ofs.close();
}

sem_t sem_main;
sem_t sem_workerstart[P];
sem_t sem_workermiddle[P];
sem_t sem_workerend[P];
mutex mtx;

void* threadFunc1(void* param) {
    threadParam_t* p = (threadParam_t*)param;
    int p_id = p->p_id;
    coo* cooM = p->cooM;
    double* pr = p->pr;
    double* updatedPr = p->updatedPr;
    int** nodeOutDegreesP = p->nodeOutDegreesP;
    sem_wait(&sem_workerstart[p_id]);//阻塞,等待主线程下发开始工作的指令
    
    //计算自己负责范围
    int range0, range1;
    range0 = N / P * p_id;
    range1 = p_id == P - 1 ? N - 1 : range0 + N / P - 1;
    //初始化矩阵M
    FILE* fp;
    int fromNodeId = 0;
    int toNodeId = 0;
    if (fopen_s(&fp, filePath, "r") != 0) { // 打开文件失败
        cout << "Thread " << p_id << " failed to open the file: " << filePath << "\n";
        exit(1);
    }
    int index = 0;
    int row = 0;
    while (fscanf_s(fp, "%d%d", &fromNodeId, &toNodeId) == 2) { //成功读取一行
        if (row >= range0 && row <= range1) {
            cooM[range0 + index].r = toNodeId;
            cooM[range0 + index].c = fromNodeId;
            cooM[range0 + index].val = 1.0;
            nodeOutDegreesP[p_id][fromNodeId]++;
            index++;
        }
        row++;
    }
    
    if (fclose(fp) != 0) {
        cout<< "Thread " << p_id << " failed to close the file: " << filePath << "\n";
        exit(1);
    }
    sem_post(&sem_main);//唤醒主线程

    sem_wait(&sem_workermiddle[p_id]); //阻塞，等待所有线程都完成自己部分的数据读写
    for (int i = range0; i <= range1; i++) {
        if (nodeOutDegrees[cooM[i].c] > 0) {
            cooM[i].val /= nodeOutDegrees[cooM[i].c];
        }
        cooM[i].val = alpha * cooM[i].val;
    }
    sem_post(&sem_main);//唤醒主线程

    //PageRank迭代
    for (int c = 0; c < C; c++) {
        sem_wait(&sem_workerend[p_id]);//阻塞，等待主线程唤醒进入下一轮
        //更新pr
        for (int i = range0; i <= range1; i++) {
            mtx.lock();
            updatedPr[cooM[i].r] += cooM[i].val * pr[cooM[i].c];
            mtx.unlock();
        }
        sem_post(&sem_main);//唤醒主线程
    }

    pthread_exit(nullptr);
}


int main(){
    //得到最大NodeId +1作为nodeN、数据行数作为COO的N
    FILE* fp;
    int fromNodeId = 0;
    int toNodeId = 0;
    int maxNodeId = 0;
    int dataRowNum = 0;
    if (fopen_s(&fp, filePath, "r") != 0) { // 打开文件失败
        cout << "Failed to open the file: " << filePath << "\n";
        exit(1);
    }
    while (fscanf_s(fp, "%d%d", &fromNodeId, &toNodeId) == 2) { //成功读取一行
        maxNodeId = max(maxNodeId, max(fromNodeId, toNodeId));
        dataRowNum++;
    }
    cout << "Data has " << dataRowNum << " rows totally.\n";
    cout << "Max NodeId is " << maxNodeId << "\n";
    N = dataRowNum;
    nodeN = maxNodeId + 1;
    if (fclose(fp) != 0) {
        cout << "Failed to close the file: " << filePath << "\n";
        exit(1);
    }
    //创建coo矩阵cooM和列向量pr，初始化pr
    coo* cooM = new coo[N];
    for (int i = 0; i < N; i++) {
        cooM[i].r = cooM[i].c = 0;
        cooM[i].val = 0.0;
    }
    double* pr = new double[nodeN];
    double* updatedPr = new double[nodeN];
    nodeOutDegrees = new int[nodeN];
    nodeOutDegreesP = new int* [P];
    for (int i = 0; i < P; i++) {
        nodeOutDegreesP[i] = new int[nodeN];
    }
    for (int i = 0; i < nodeN; i++) {
        pr[i] = 1 / (double)nodeN;
        updatedPr[i] = (1 - alpha) / nodeN * pr[i];
        //初始化出度统计向量nodeOutDegrees
        nodeOutDegrees[i] = 0;
        for (int p = 0; p < P; p++) {
            nodeOutDegreesP[p][i] = 0;
        }
    }

    //初始化信号量
    sem_init(&sem_main, 0, 0);
    for (int i = 0; i < P; i++) {
        sem_init(&sem_workerstart[i], 0, 0);
        sem_init(&sem_workermiddle[i], 0, 0);
        sem_init(&sem_workerend[i], 0, 0);
    }
    //创建线程
    pthread_t handles[P];
    threadParam_t param[P];
    for (int p_id = 0; p_id < P; p_id++) {
        param[p_id].p_id = p_id;
        param[p_id].cooM = cooM;
        param[p_id].pr = pr;
        param[p_id].updatedPr = updatedPr;
        param[p_id].nodeOutDegreesP = nodeOutDegreesP;
        pthread_create(handles + p_id, nullptr, threadFunc1, param + p_id);
    }

    //初始化矩阵M
    //唤醒工作线程 && 主线程睡眠
    for (int p_id = 0; p_id < P; p_id++) {
        sem_post(&sem_workerstart[p_id]);
    }
    for (int p_id = 0; p_id < P; p_id++) {
        sem_wait(&sem_main);
    }
    //完成出度统计向量
    for (int i = 0; i < nodeN; i++) {
        for (int p = 0; p < P; p++) {
            nodeOutDegrees[i] += nodeOutDegreesP[p][i];
        }
    }
    //唤醒工作线程 && 主线程睡眠
    for (int p_id = 0; p_id < P; p_id++) {
        sem_post(&sem_workermiddle[p_id]);
    }
    for (int p_id = 0; p_id < P; p_id++) {
        sem_wait(&sem_main);
    }

    //PageRank迭代
    cout << "Start PageRank.\n";
    prRes* checkRes = new prRes[nodeN];
    prRes* lastCheckRes = new prRes[nodeN]; //用于收敛判断
    for (int c = 0; c < C; c++) {
        //主线程工作
        cout << "Round " << c + 1 << " ...\n";
        //主线程唤醒工作线程 && 主线程睡眠
        for (int p_id = 0; p_id < P; p_id++) {
            sem_post(&sem_workerend[p_id]);
        }
        for (int p_id = 0; p_id < P; p_id++) {
            sem_wait(&sem_main);
        }
        for (int i = 0; i < nodeN; i++) {
            pr[i] = updatedPr[i];
            updatedPr[i] = (1 - alpha) / nodeN * pr[i];
        }
        //判断收敛与否
        if (c == 0) {
            for (int i = 0; i < nodeN; i++) {
                lastCheckRes[i].nodeId = i;
                lastCheckRes[i].prVal = pr[i];
            }
        }
        int diff = 0;
        for (int i = 0; i < nodeN; i++) {
            checkRes[i].nodeId = i;
            checkRes[i].prVal = pr[i];
        }
        sort(checkRes, checkRes + nodeN, cmp);
        for (int i = 0; i < nodeN; i++) {
            if (lastCheckRes[i].nodeId != checkRes[i].nodeId) {
                diff++;
            }
        }
        for (int i = 0; i < nodeN; i++) {
            lastCheckRes[i].nodeId = checkRes[i].nodeId;
            lastCheckRes[i].prVal = checkRes[i].nodeId;
        }
        if (double(diff) > 0 * double(nodeN)) { //不收敛
            C++;
        }
    }

    //等待线程执行结束
    for (int p_id = 0; p_id < P; p_id++) {
        pthread_join(handles[p_id], nullptr);
    }
    //销毁信号量
    sem_destroy(&sem_main);
    for (int i = 0; i < P; i++) {
        sem_destroy(&sem_workerstart[i]);
        sem_destroy(&sem_workermiddle[i]);
        sem_destroy(&sem_workerend[i]);
    }
    
    //pagerank结果展示
    prRes* result = new prRes[nodeN];
    for (int i = 0; i < nodeN; i++) {
        result[i].nodeId = i;
        result[i].prVal = pr[i];
    }
    displayPr(result);
    //displayM(cooM);
    
    //写文件
    saveResult(result);
    
    cout << "End PageRank.\n";
    return 0;
}